package com.aotain.project.apollo.bolt;

import java.text.SimpleDateFormat;
//import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
//import java.util.Map.Entry;




//import org.apache.commons.codec.binary.Base64;
import org.apache.log4j.Logger;
import org.apache.storm.Config;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.FailedException;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Tuple;

import com.aotain.hbase.HBaseRecordAdd;
import com.aotain.apollo.ApolloConfig;
import com.aotain.common.CommonFunction;
import com.aotain.storm.AbstractBolt;

/**
 * 根据 域名和 机房统计 PV值
 * 2016/02/23 新建
 * @author 胡德意
 *
 */
public class IDCDomainStatBolt extends AbstractBolt{

	/**
	 * 根域名PV统计
	 */
	
	OutputCollector _collector;
	
	//HashMap<String,ServerInfo> _mapServers = null;
	
	String IDCDomainMapKey = null;
	Map<String,Integer> IDCDomainMap = new HashMap<String,Integer>();
	
	//long TotalPV = 0L;


	
	private void countInLocal(Tuple tuple) {

		if(tuple != null ) {  
			//System.out.println("..........IDCDomainStatBolt......... " + System.currentTimeMillis());
			
			String domainname = tuple.getStringByField("domainname");
			if(domainname==null || domainname.trim().isEmpty())
				return;

			String domain = CommonFunction.getRootDomain(domainname);
			if(domain==null || domain.trim().isEmpty()){
				//System.out.println("domainname is " + domainname +",domain is " + domain);
				return;}
			
			String idcno = tuple.getStringByField("idcno");			
			IDCDomainMapKey = idcno + "_"  + domain;

			int IDCDomainPV = 1;
			if(IDCDomainMap.get(IDCDomainMapKey) != null) {
				IDCDomainPV = IDCDomainMap.get(IDCDomainMapKey) + 1;
			}
			//System.out.println("domainname is " + domainname +",domain is " + domain +", idc is "+ idcno+", pv is "+ IDCDomainPV);
			IDCDomainMap.put(IDCDomainMapKey, IDCDomainPV);			
			
		} else {  
			System.err.println(" >>>>>>>>>>>>>>>>IDCDomainStatBolt tuple is null <<<<<<<<<<<<<<<");
		}
	}
	
	private void emitCountingData() {
		
		Date date = new Date();
		SimpleDateFormat sdf_d  = new SimpleDateFormat("yyyyMMdd");
		SimpleDateFormat sdfm = new SimpleDateFormat("yyyyMMddHHmm00");
		String dateStr_m = sdfm.format(date);
		String dateStr_d = sdf_d.format(date);
	
		HBaseRecordAdd hbaseInstance = HBaseRecordAdd.getInstance(
				zooserver, logserver, brokerList);
		
		for(String dkey : IDCDomainMap.keySet()) {
			
			int avgPvalue = IDCDomainMap.get(dkey);			
			String cf = "cf";
            String Domain=dkey.substring(dkey.lastIndexOf('_')+1);
			String rowkeyDay = String.format("%s_%s", dateStr_d,dkey);
			
			//System.out.println("rowkeyDay is " + rowkeyDay +",dateStr_m is " + dateStr_m +",Domain is " + Domain +", pv is "+ avgPvalue);
			
			hbaseInstance.Add("SDS_DOMAIN_PV_D", rowkeyDay, cf,  "DOMAIN", Domain);
			hbaseInstance.Add("SDS_DOMAIN_PV_D", rowkeyDay, cf,  "REPORTTIME", dateStr_m);
			hbaseInstance.incerment("SDS_DOMAIN_PV_D", rowkeyDay, "cf:PV", avgPvalue);
			
		}
		
		IDCDomainMap.clear();
		
		
		
	}
	
	@Override
	public void execute(Tuple input) {
		// TODO Auto-generated method stub
		try {
			
			if(isTickTuple(input)) {//满足时间条件
				emitCountingData();  //计算，入库
				_collector.ack(input);
			} else {//不满足时间条件
				countInLocal(input); //数据记录 
				_collector.ack(input);
			}
		} catch (Exception e) {
			// TODO Auto-generated catch block
			Logger.getRootLogger().error(e);
			throw new FailedException("IDCDomainStatBolt Exception",e);
		}
	}

	@Override
	public void cleanup() {
		// TODO Auto-generated method stub
		
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		// TODO Auto-generated method stub
		
	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
		// TODO Auto-generated method stub
		Map<String,Object> conf = new HashMap<String,Object>();
		conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS,1800);
		return conf;
	}


	@Override
	public void Init(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		// TODO Auto-generated method stub
		_collector = collector;
		String dbJson = stormConf.get("dbjson").toString();
		ApolloConfig ap = new ApolloConfig(dbJson);
	}

}
